package com.pipe.connector.actor;

import com.pipe.connector.RowDataListener;
import com.pipe.exception.ColumnReaderException;
import com.pipe.exception.ProtocolCodecException;
import com.pipe.exception.SessionException;
import com.pipe.protocol.BasePackage;
import com.pipe.protocol.events.BinLogEventHeader;
import com.pipe.protocol.events.other.FormatDescriptionEvent;
import com.pipe.protocol.events.other.RotateEvent;
import com.pipe.protocol.events.table.TableMapEvent;
import com.pipe.protocol.events.table.TableMapEventPayLoad;
import com.pipe.protocol.events.table.TableMapEventPostHeader;
import com.pipe.protocol.events.write.WriteRowEventV2;
import com.pipe.protocol.events.write.WriteRowEventV2Body;
import com.pipe.protocol.events.write.WriteRowEventV2Header;
import com.pipe.protocol.events.write.WriteRowEventV2Rows;
import com.pipe.protocol.login.OkAndEofPacket;
import com.pipe.protocol.query.TableMetaInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

/**
 * @author Kiven
 */
public class BinlogEventActor {
    private static Logger LOG = LoggerFactory.getLogger(BinlogEventActor.class);
    private RowDataListener listener;

    public BinlogEventActor(RowDataListener listener) {
        this.listener = listener;
    }

    TableMapEvent handler(BasePackage basePackage, TableMapEvent tableMapEvent, Map<String, TableMetaInfo> tableMetaInfoMap) throws SessionException, ProtocolCodecException {
        OkAndEofPacket okAndEofPacket = OkAndEofPacket.shortBuild(basePackage.getPayLoadByteBuf());
        if (okAndEofPacket.isOkHead() && basePackage.getPayLoadLen() >= OkAndEofPacket.OK_PACKAGE_LENGTH) {
            BinLogEventHeader binLogEventHeader = BinLogEventHeader.build(basePackage.getPayLoadByteBuf());
            if (binLogEventHeader.isTableMapEvent()) {
                TableMapEventPostHeader postHeader = TableMapEventPostHeader.build(basePackage.getPayLoadByteBuf());
                TableMapEventPayLoad payLoad = TableMapEventPayLoad.build(basePackage.getPayLoadByteBuf());
                TableMapEvent event = new TableMapEvent(postHeader, payLoad);
                LOG.info("[BINLOG EVENT] [TableMapEvent] - received!");
                LOG.info("[BINLOG EVENT] [TableMapEvent] - " + event.toString());
                return event;
            }
            if (binLogEventHeader.isWriteRowsEventV2() || binLogEventHeader.isUpdateRowsEventV2()) {
                WriteRowEventV2Header header = WriteRowEventV2Header.build(basePackage.getPayLoadByteBuf());
                WriteRowEventV2Body body = WriteRowEventV2Body.build(basePackage.getPayLoadByteBuf(), binLogEventHeader.isUpdateRowsEventV2());
                WriteRowEventV2Rows rows = null;
                try {
                    rows = WriteRowEventV2Rows.build(basePackage.getPayLoadByteBuf(), (int) body.getColumnNum(), tableMapEvent, tableMetaInfoMap, binLogEventHeader.isUpdateRowsEventV2());
                } catch (ColumnReaderException e) {
                    throw new ProtocolCodecException("[BINLOG EVENT] [WriteRowEventV2] - ColumnReader exception!", e);
                }
                WriteRowEventV2 writeRowEventV2 = new WriteRowEventV2(header, body, rows);
                LOG.info("[BINLOG EVENT] [WriteRowEventV2] - received!");
                LOG.info("[BINLOG EVENT] [WriteRowEventV2] - {}", writeRowEventV2.toString());
                listener.newDataArrived(rows);
                return null;
            }
            if (binLogEventHeader.isRotateEvent()) {
                RotateEvent rotateEvent = RotateEvent.build(basePackage.getPayLoadByteBuf());
                LOG.info("[BINLOG EVENT] [RotateEvent] - received!");
                LOG.info("[BINLOG EVENT] [RotateEvent] - {}", rotateEvent.toString());
                return null;
            }
            if (binLogEventHeader.isFormatDescriptionEvent()) {
                FormatDescriptionEvent descriptionEvent = FormatDescriptionEvent.build(basePackage.getPayLoadByteBuf());
                LOG.info("[BINLOG EVENT] [FormatDescriptionEvent] - received!");
                LOG.info("[BINLOG EVENT] [FormatDescriptionEvent] - {}", descriptionEvent.toString());
                return null;
            }
            LOG.info("[BINLOG EVENT] [UnknownEvent] - received!");
            return null;
        }
        if (okAndEofPacket.isEofHead() && basePackage.getPayLoadLen() < OkAndEofPacket.EOF_PACKAGE_LENGTH) {
            LOG.info("[BINLOG EVENT] - Eof package received! Dumping failed!");
            throw new SessionException("[BINLOG EVENT] - Eof package received! Dumping failed!");
        }
        LOG.info("[BINLOG EVENT] - Package of unknown type recieved! Dumping failed!");
        throw new SessionException("[BINLOG EVENT] - Package of unknown type recieved! Dumping failed!");
    }
}